package com.bwie.shop.mq;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.bwie.shop.dao.ShopDao;
import com.bwie.shop.entity.ShopEntity;
import com.rabbitmq.client.Channel;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;

@Component
public class Customr {
    @Autowired
    private ShopDao shopDao;
    @Autowired
    private RestHighLevelClient client;
    @Autowired
    private RedisTemplate redisTemplate;
   @Autowired
   private ThreadPoolExecutor executor;
   //②新增：消息队列名称：merchandise_new （10分）
    @RabbitListener(queues = "merchandise_new")
    @RabbitHandler
    public void getMsg1(String msg, Message message, Channel channel) throws IOException {

        System.out.println("添加"+msg);
        ShopEntity shopEntity = JSON.parseObject(msg, new TypeReference<ShopEntity>() {
        });
        shopDao.insert(shopEntity);

        executor.submit(()->{
//商品搜索服务接受到消息之后，使用线程池进行异步增量ES搜素服务数据（6分）
//使用商品ID调用商品维护服务的商品详情接口（2分）

//查询到的商品信息进行ES增量同步（2分）

            System.out.println("-------查询到的商品信息进行ES增量同步----------");
            List<ShopEntity> list2 = shopDao.selectList(null);
            if (list2!=null){
                BulkRequest request = new BulkRequest();
                for (ShopEntity shopEntity1 : list2) {
                    IndexRequest request2 = new IndexRequest("lastmonth");
                    request2.id(shopEntity1.getSid()+"");
                    String jsonString = JSON.toJSONString(shopEntity1);
                    request2.source(jsonString, XContentType.JSON);
                    request.add(request2);
                }
                try {
                    BulkResponse bulkResponse = client.bulk(request, RequestOptions.DEFAULT);
                } catch (IOException e) {
                    e.printStackTrace();
                }

            }
//商品库存添加到Redis当中（2分)
            for (ShopEntity entity : list2) {
                BoundHashOperations shop = redisTemplate.boundHashOps("shop");
                if (shop.get(entity.getSname())==null){

                    shop.put(entity.getSname(),entity.getCount());
                    System.out.println("商品库存添加到Redis当中"+entity.getSname()+entity.getCount());
                }

            }


        });
        //①MQ必须保证消息可靠性（2分）
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

    }
    @RabbitListener(queues = "merchandise_update")
    public void getMsg2(String msg,Message message,Channel channel) throws IOException {
        System.out.println("修改"+msg);
        ShopEntity shopEntity = JSON.parseObject(msg, new TypeReference<ShopEntity>() {
        });
        shopDao.updateById(shopEntity);
        executor.submit(()->{
            System.out.println("-------刷新ES当中的商品信息----------");
        List<ShopEntity> list2 = shopDao.selectList(null);
        if (list2!=null){
            BulkRequest request = new BulkRequest();
            for (ShopEntity shopEntity1 : list2) {
                IndexRequest request2 = new IndexRequest("lastmonth");
                request2.id(shopEntity1.getSid()+"");
                String jsonString = JSON.toJSONString(shopEntity1);
                request2.source(jsonString, XContentType.JSON);
                request.add(request2);
            }
            System.out.println(list2);
            try {
                BulkResponse bulkResponse = client.bulk(request, RequestOptions.DEFAULT);
            } catch (IOException e) {
                e.printStackTrace();
            }

        }
            for (ShopEntity entity : list2) {
                BoundHashOperations shop = redisTemplate.boundHashOps("shop");
                if (shop.get(entity.getSname())!=null){

                    shop.put(entity.getSname(),entity.getCount());
                    System.out.println("刷新Redis当中商品的库存量"+entity.getSname()+entity.getCount());
                }

            }
        });
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);


    }

    @RabbitListener(queues = "merchandise_on_shelf")
    public void getMsg3(String msg,Message message,Channel channel) throws IOException {
        System.out.println("启用"+msg);
        ShopEntity shopEntity = JSON.parseObject(msg, new TypeReference<ShopEntity>() {
        });
        shopDao.qi(shopEntity);
        executor.submit(()->{
            System.out.println("-------刷新ES当中的商品信息----------");
            List<ShopEntity> list2 = shopDao.selectList(null);
            if (list2!=null){
                BulkRequest request = new BulkRequest();
                for (ShopEntity shopEntity1 : list2) {
                    IndexRequest request2 = new IndexRequest("lastmonth");
                    request2.id(shopEntity1.getSid()+"");
                    String jsonString = JSON.toJSONString(shopEntity1);
                    request2.source(jsonString, XContentType.JSON);
                    request.add(request2);
                }
                System.out.println(list2);
                try {
                    BulkResponse bulkResponse = client.bulk(request, RequestOptions.DEFAULT);
                } catch (IOException e) {
                    e.printStackTrace();
                }

            }
        });
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);


    }
    @RabbitListener(queues = "merchandise_down_shelf")
    public void getMsg4(String msg,Message message,Channel channel) throws IOException {
        System.out.println("禁用"+msg);
        ShopEntity shopEntity = JSON.parseObject(msg, new TypeReference<ShopEntity>() {
        });
        shopDao.jin(shopEntity);
        executor.submit(()->{
            System.out.println("-------刷新ES当中的商品信息----------");
            List<ShopEntity> list2 = shopDao.selectList(null);
            if (list2!=null){
                BulkRequest request = new BulkRequest();
                for (ShopEntity shopEntity1 : list2) {
                    IndexRequest request2 = new IndexRequest("lastmonth");
                    request2.id(shopEntity1.getSid()+"");
                    String jsonString = JSON.toJSONString(shopEntity1);
                    request2.source(jsonString, XContentType.JSON);
                    request.add(request2);
                }
                System.out.println(list2);
                try {
                    BulkResponse bulkResponse = client.bulk(request, RequestOptions.DEFAULT);
                } catch (IOException e) {
                    e.printStackTrace();
                }

            }
        });
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);


    }
    @RabbitListener(queues = "merchandise_delete")
    public void getMsg5(Integer[] sids,Message message,Channel channel) throws IOException {
        System.out.println("删除id:"+Arrays.asList(sids));

        shopDao.deleteBatchIds(Arrays.asList(sids));
        executor.submit(()->{
            System.out.println("-------刷新ES当中的商品信息----------");
            List<ShopEntity> list2 = shopDao.selectList(null);
            if (list2!=null){
                BulkRequest request = new BulkRequest();
                for (ShopEntity shopEntity1 : list2) {
                    IndexRequest request2 = new IndexRequest("lastmonth");
                    request2.id(shopEntity1.getSid()+"");
                    String jsonString = JSON.toJSONString(shopEntity1);
                    request2.source(jsonString, XContentType.JSON);
                    request.add(request2);
                }
                System.out.println(list2);
                try {
                    BulkResponse bulkResponse = client.bulk(request, RequestOptions.DEFAULT);
                } catch (IOException e) {
                    e.printStackTrace();
                }

            }
        });
        channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);


    }

}
